package com.erainm.customserializer;

import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Serializer;

import java.nio.ByteBuffer;
import java.util.Map;

/**
 * Created by erainm on 2021/9/9 4:13 下午.
 *
 * @Description TODO:自定义序列化
 **/
public class UserSerializer implements Serializer<User> {
    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        // 用于接收对序列化器的配置参数，并对当前序列化器进行配置和初始化
    }

    @Override
    public byte[] serialize(String topic, User data) {
        try {
            if (data == null) {
                return null;
            } else {
                Integer userId = data.getUserId();
                String username = data.getUsername();

                if (userId != null) {
                    if (username != null) {

                        byte[] bytes = username.getBytes("utf-8");

                        int lenth = bytes.length;
                        // 第一个4 ：4个字节，用于存储userId的值
                        // 第二个4：4个字节，用于存储username字节数据的长度int值
                        // 第三个长度，用于存放username序列化之后的字节数组
                        ByteBuffer buffer = ByteBuffer.allocate(4 + 4 + lenth);
                        // 设置userId
                        buffer.putInt(userId);
                        // 设置username字节数组长度
                        buffer.putInt(lenth);
                        // 设置username字节数组
                        buffer.put(bytes);
                        // 以字节数组形式返回user对象值
                        return buffer.array();
                    }
                }
            }
        } catch (Exception e) {
            throw new SerializationException("序列化失败！");
        }
        return null;
    }

    @Override
    public void close() {
        // 用于关闭资源等操作，需要幂等，即多次调用，效果是一样的
    }
}
